package com.xnyzc.lhy.task.mq;

import com.alibaba.fastjson.JSON;
import com.xnyzc.lhy.common.component.config.security.RocketConfig;
import com.xnyzc.lhy.common.constant.MqConstant;
import com.xnyzc.lhy.common.constant.MqConsumerGroupConstant;
import com.xnyzc.lhy.common.util.CheckUtil;
import com.xnyzc.lhy.task.entity.demo.CharterOrderInfoCreated;
import com.xnyzc.lhy.task.entity.statistical.CharterStatisticalData;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.regex.Pattern;

@Slf4j
@Component
public class MQConsumerForCharterCreateOrder {
    /**
     * rocketMq 配置文件
     */

    private DefaultMQPushConsumer consumer;

    @Autowired
    RocketConfig rocketConfig;

    @Autowired
    private MongoTemplate mongoTemplate;

//    @Value("${rocketMq.isStart}")
//    private boolean isStart;

    /**
     * 开启监听
     */
    @PostConstruct
    public void init() {
        log.info(rocketConfig.getTopic() + "--------------------------------------------------" + rocketConfig.getNameServerAddr());
        try {
            consumer = new DefaultMQPushConsumer(MqConsumerGroupConstant.CHARTER_ORDER_CREATED);
            consumer.setNamesrvAddr(rocketConfig.getNameServerAddr());
            consumer.subscribe(rocketConfig.getTopic(), MqConstant.CHARTER_ORDER_CREATED);
            //批量消费的最大消息条数
            consumer.setConsumeMessageBatchMaxSize(10);
            //设置消费策略:队列头消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            /**
             *  设置监听
             */
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        MQConsumerForCharterCreateOrder.this.dealBody(msg);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            log.info("---------MQConsumer启动完成------------");
        } catch (Exception e) {
            e.printStackTrace();
            log.info("---------MQConsumer初始化失败------------",e);
        }
    }

    private void dealBody(MessageExt messageExt) {
        String data = new String(messageExt.getBody());
        log.info("订单生成的个别数据：" + data);
        CharterOrderInfoCreated orderInfoCreated = JSON.parseObject(data, CharterOrderInfoCreated.class);
        if(CheckUtil.objIsEmpty(orderInfoCreated)){
            return;
        }
        String priceMarker = orderInfoCreated.getPriceMarker();
        if(CheckUtil.objIsEmpty(priceMarker)){
            return;
        }
        //根据价格标志跟新mongo库中的数据
        Update update = new Update().set("isDeal", 1);
        update.set("orderCreateTime",orderInfoCreated.getDealTime());
        update.set("vehicleType",orderInfoCreated.getVehicleType());
        update.set("comboType", orderInfoCreated.getComboType());
        update.set("priceMarker", priceMarker);
        log.info("开始跟新数据："+orderInfoCreated);
        Pattern pattern=Pattern.compile("^.*"+priceMarker+".*$", Pattern.CASE_INSENSITIVE);
        mongoTemplate.upsert(new Query(Criteria.where("price").regex(pattern)), update,
                           CharterStatisticalData.class,"charterStatisticalData");
    }

}
